from pyspark.sql import Window
from pyspark.sql.session import SparkSession
from pyspark.sql.functions import *
spark= SparkSession.builder.appName('demo1_sparksession').enableHiveSupport().config('spark.sql.shuffle.partitions', 2).config("spark.jars", "jars/mysql-connector-java-8.0.29.jar").getOrCreate()
# /user/root/dm_db

# | 字段名        | 数据类型 | 描述      |
# | ------------- | -------- | --------- |
# | prov_id       | STRING   | 省份ID    |
# | prov_name     | STRING   | 省份名称  |
# | city_id       | STRING   | 城市ID    |
# | city_name     | STRING   | 城市名称  |
# | county_id     | STRING   | 区县ID    |
# | county_name   | STRING   | 区县名称  |
# | city_level    | STRING   | 城市等级  |
# | economic_belt | STRING   | 经济带    |
# | city_feature1 | STRING   | 城市特征1 |
staypoint_df = spark.read.format('csv').option('sep','\t').schema('mdn string,date string,county string,lon double,lat double,bsid string,grid_id string,biz_type string,event_type string,data_source string')\
    .load('D:\\bigdata35\\data\\staypoint.txt')

usertag_df = spark.read.format('csv').option('sep',',').schema('mdn string,name string,gender string,age string,id_number string,number_attr string,trmnl_brand string,trmnl_price string,packg string,conpot string,resi_grid_id string,resi_county_id string')\
    .load('D:\\bigdata35\\data\\usertag.txt')

admincode_df = spark.read.format('csv').option('sep',',').schema('city_id string,city_name string,prov_id string,prov_name string,county_id string,county_name string,city_level string,economic_belt string,city_feature1 string')\
    .load('D:\\bigdata35\\data\\admincode.txt')

# ##### 3.1 用户移动模式聚类（15分）
#
# - **分析目标**：识别典型的用户移动模式
# - **分析方法**：
#   - 基于停留点序列的聚类分析
#   - 典型模式如：
#     - 多区域活动型

staypoint_df = staypoint_df.join(usertag_df, 'mdn', 'inner')\
    .withColumn('area',collect_set('county').over(Window.partitionBy('mdn').orderBy('date')))\
    .withColumn('area_len',size('area'))\
    .withColumn('act_flag',when(col('area_len')>2,1).otherwise(0))


staypoint_df = staypoint_df.filter('act_flag == 1') \
    .select('mdn', 'area') \
    .distinct() \
    .withColumn('area_str', concat_ws(',', col('area')))



staypoint_df.select('mdn', 'area_str') \
    .write \
    .format("jdbc") \
    .option("url", "jdbc:mysql://master:3306/dianxim") \
    .option("driver", "com.mysql.cj.jdbc.Driver") \
    .option("dbtable", "demo_5") \
    .option("user", "root") \
    .option("password", "123456") \
    .mode("overwrite") \
    .save()

























